package com.cloud.shopping.mq;

import com.alibaba.fastjson.JSON;
import com.cloud.shopping.constants.RocketConstants;
import com.cloud.shopping.order.service.OrderService;
import com.cloud.shopping.pay.model.PayLog;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;

/**
 * @author John.zhang
 * @version Id: OrderResultListener, v 1.0 2021/1/5 11:27 ZhiYong Exp $
 */

@Component
@RocketMQMessageListener(topic = RocketConstants.PAY_LOG_TOPIC, consumerGroup = RocketConstants.CONSUMER_GROUP)
public class OrderResultListener implements RocketMQListener, RocketMQPushConsumerLifecycleListener {

    @Resource
    private OrderService orderService;

    @Override
    public void onMessage(Object o) {

    }

    /**
     * 消息监听
     *
     * @param defaultMQPushConsumer defaultMQPushConsumer
     */
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
            for (MessageExt msg : messages) {
                String result = new String(msg.getBody(), StandardCharsets.UTF_8);
                PayLog payLog = JSON.parseObject(result, PayLog.class);
                if (payLog.getStatus() == 2) {
                    //支付成功
                    int count = orderService.updateAfterPayStatus(payLog.getPayId());
                    System.out.println(payLog.getId() + "================" + count);
                } else {
                    //支付失败
                    //1：修改订单状态
                    //2：库存回滚
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
    }
}
